package com.yuandian.ops.workbench.controller;

import com.alibaba.fastjson.JSON;
import com.yuandian.bpm.common.core.util.R;
import com.yuandian.bpm.common.core.util.SpringContextHolder;
import com.yuandian.bpm.common.security.annotation.Inner;
import com.yuandian.bpm.common.websocket.distribute.MessageDO;
import com.yuandian.bpm.common.websocket.distribute.RedisMessageDistributor;
import com.yuandian.ops.workbench.vo.BusinessPushDataVo;
import com.yuandian.utils.YDateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.Date;

/**
 * 说明：
 *
 * @Author zfsn
 * @Date 2023/2/15 19:14
 */
@RequestMapping("/ws")
@RestController
@Slf4j
public class WsController {

	@Inner(value = false)
	@GetMapping("/send")
	public R send() {
		RedisMessageDistributor messageDistributor = SpringContextHolder.getBean(RedisMessageDistributor.class);
		String message = "content-" + System.currentTimeMillis();
		// 发给一个人
		// messageDistributor.distribute(MessageDO.singleMessage(1, message));
		// 发给多个人
		// messageDistributor.distribute(MessageDO.multipleMessage(CollUtil.newArrayList(1, 3, 4, 8), message));
		// 发给所有人
		messageDistributor.distribute(MessageDO.broadcastMessage(message));
		return R.ok();
	}



	/**
	 * kafka消费者
	 * @param record 消费信息
	 */
	@KafkaListener(topics = "#{'${spring.kafka.topic-prefix}' + 'topic_business_push_alarm'}")
	public void listenOpsAlarm(ConsumerRecord<String, String> record) {
		BusinessPushDataVo vo = JSON.parseObject(record.value(), BusinessPushDataVo.class);
		log.info("运维工作台-消费Kafka数据: {}", vo);

		RedisMessageDistributor messageDistributor = SpringContextHolder.getBean(RedisMessageDistributor.class);
		// 发给一个人
		// messageDistributor.distribute(MessageDO.singleMessage(1, message));
		// 发给多个人
		// messageDistributor.distribute(MessageDO.multipleMessage(CollUtil.newArrayList(1, 3, 4, 8), message));
		// 发给所有人
		MessageDO messageDO = new MessageDO();
		messageDO.setNeedBroadcast(Boolean.TRUE);
		messageDO.setSessionKeys(Arrays.asList(1, 1));
		messageDO.setMessageText(JSON.toJSONString(R.ok(vo)));
//		messageDistributor.distribute(messageDO);
	}

	@GetMapping("sendTest")
	public Object sendTest() {
		BusinessPushDataVo pushData = new BusinessPushDataVo();
		pushData.setGroupId("12323234123235L");
		pushData.setStepCode("b1282d97b4e5405c");
		pushData.setStepName("测试步骤");
		pushData.setStartTime("2023-01-29 00:00:00");
		pushData.setEndTime("2023-01-29 00:00:00");
		pushData.setFailRate(35.91D);
		pushData.setDealTimeAlarmRate(65.66D);
		pushData.setFileType("BS_FAIL_RATE_10");
		pushData.setPushTime("2023-01-29 00:00:00");

		pushData.setSystemId(1);
		pushData.setSystemName("测试系统");
		pushData.setBusinessCode("BR-SB-00000102");
		pushData.setBusinessName("测试业务");
		pushData.setIsTop(1);
		pushData.setDataType(5);
		pushData.setRecords(3);

//        senderService.send(JSON.toJSONString(Result.setSuccess(pushData)));

		RedisMessageDistributor messageDistributor = SpringContextHolder.getBean(RedisMessageDistributor.class);
		// websocket 发送消息
		MessageDO messageDO = new MessageDO();
		messageDO.setNeedBroadcast(Boolean.TRUE);
		messageDO.setSessionKeys(Arrays.asList(1, 1));
		messageDO.setMessageText(JSON.toJSONString(R.ok(pushData)));
//		messageDistributor.distribute(messageDO);

		log.info("运维工作台推送数据: {}", pushData);

		return "OK";
	}


//	@Scheduled(cron = "*/5 * * * * ?")
	public void testSend() {

		BusinessPushDataVo pushData = new BusinessPushDataVo();
		pushData.setGroupId(YDateUtil.second(new Date()) + "");
		pushData.setStepCode("b1282d97b4e5405c");
		pushData.setStepName("测试步骤");
		pushData.setStartTime("2023-01-29 00:00:00");
		pushData.setEndTime("2023-01-29 00:00:00");
		pushData.setFailRate(35.91D);
		pushData.setDealTimeAlarmRate(65.66D);
		pushData.setFileType("BS_FAIL_RATE_10");
		pushData.setPushTime("2023-01-29 00:00:00");

		pushData.setSystemId(1);
		pushData.setSystemName("测试系统");
		pushData.setBusinessCode(YDateUtil.second(new Date()) + "");
		pushData.setBusinessName(YDateUtil.second(new Date()) + "");
		pushData.setIsTop(1);
		pushData.setDataType(5);
		pushData.setRecords(3);

		RedisMessageDistributor messageDistributor = SpringContextHolder.getBean(RedisMessageDistributor.class);
		// websocket 发送消息
		MessageDO messageDO = new MessageDO();
		messageDO.setNeedBroadcast(Boolean.TRUE);
		messageDO.setSessionKeys(Arrays.asList(1, 1));
		messageDO.setMessageText(JSON.toJSONString(R.ok(pushData)));
		messageDistributor.distribute(messageDO);
	}
}
